package com.renrendoc.network.flow.ana;

import com.renrendoc.network.flow.beans.NginxLogEvent;
import com.renrendoc.network.flow.beans.out.PageEvent;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.tuple.Tuple4;

import java.text.DecimalFormat;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;

/**
 * 页面pv,uv,iv 聚合操作
 */
public class PageAnaAgg implements AggregateFunction<NginxLogEvent, Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>>, PageEvent> {
    @Override
    public Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> createAccumulator() {
        return Tuple4.of(0L, new HashSet<>(), new HashSet<>(), new HashMap<>());
    }

    @Override
    public Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> add(NginxLogEvent nginxLogEvent, Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> t) {
        t.f1.add(nginxLogEvent.ci); // uv set
        t.f2.add(nginxLogEvent.ip); // iv set 记得修改成最终的ip数据信息
        Long value = t.f3.get(nginxLogEvent.si);
        if (null == t.f3.get(nginxLogEvent.si)) {
            t.f3.put(nginxLogEvent.si, 1L);
        } else {
            t.f3.put(nginxLogEvent.si, value + 1);
        }
        return Tuple4.of(t.f0 + 1, t.f1, t.f2, t.f3);
    }

    @Override
    public PageEvent getResult(Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> t) {

        int one = 1;
        int total = 1;
        for (Map.Entry<String, Long> entry : t.f3.entrySet()) {
            Long value = entry.getValue();
            if (1 == value) {
                one += 1;
            }
            total +=1 ;
        }

        // 创建 DecimalFormat 对象，并设置格式为保留两位小数
        DecimalFormat df = new DecimalFormat("#.##");
        // 格式化 double 数值
        double jump = Double.parseDouble(df.format(one / total));

        return new PageEvent("", t.f0,
                t.f1.size(),
                t.f2.size(),
                0,
                jump
        );
    }

    @Override
    public Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> merge(Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> longHashSetHashSetTuple3, Tuple4<Long, HashSet<String>, HashSet<String>, HashMap<String, Long>> acc1) {
        return null;
    }
}
